<?php

namespace app\common\service\GoodsService;

use think\Db;
use app\Normal;
use app\model\BbcLogModel\Sync;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Exchange\AMQPExchangeType;


/**
 * ES数据队列
 * @author linan
 */
class ChannelEsService
{
    public $queueName=[
        'channel'=>'document_qqc_qqccommodity',
        'store'   =>'document_qd_qdcommodity',
    ];
    public $time;

    /**
     * 写入队列
     * @param array $data
     * @param string $type
     * @param string $method
     * @return  array
     */
    function pushNew($data,$type,$method){
        $account=config('amqp');
        $host = $account['host'];
        $port = $account['port'];
        $username = $account['username'];
        $password = $account['password'];
        $exchange_name= $this->queueName[$type];
        $queueName = $this->queueName[$type];
        $routing_key= $this->queueName[$type];
        $connection=new AMQPStreamConnection($host, $port, $username,$password,'/',false,'AMQPLAIN',null,'en_US',60.0,60.0,null,true,30,60.0,null);
        $channel=$connection->channel();
        $channel->queue_declare($queueName, false, true, false, false);
        $channel->exchange_declare($exchange_name,AMQPExchangeType::DIRECT,false,true,false);
        $channel->queue_bind(
            $queueName, // 队列名
            $exchange_name, // 交换机名
            $routing_key // 路由key
        );
        if($method=='delete'){
            $actionType=0;
        }else{
            $actionType=1;
        }
        $num=count($data);
        if($num>100){
            $flag=true;
            $i=0;
            while ($flag){
                $new=array_slice($data,$i,100);
                if($new){
                    $msg=new AMQPMessage(json_encode(['actionType'=>$actionType,'datas'=>$new],JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
                    $channel->basic_publish($msg,false,$routing_key);
                    $i+=100;
                }else{
                    $flag=false;
                }
            }
        }else{
            $msg=new AMQPMessage(json_encode(['actionType'=>$actionType,'datas'=>$data],JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
            $channel->basic_publish($msg,false,$routing_key);
        }

        $channel->close();
        $connection->close();
        return  ['msg'=>'ok'];
    }

}

